/*
 * Copyright 2003-2007 (c) Live Media Pty Ltd. <argot@einet.com.au> 
 *
 * This software is licensed under the Argot Public License 
 * which may be found in the file LICENSE distributed 
 * with this software.
 *
 * More information about this license can be found at
 * http://www.einet.com.au/License
 * 
 * The Developer of this software is Live Media Pty Ltd,
 * PO Box 4591, Melbourne 3001, Australia.  The license is subject 
 * to the law of Victoria, Australia, and subject to exclusive 
 * jurisdiction of the Victorian courts.
 */

package com.colony.concurrent;

/**
 *  A Queue class which extends the simple Linked List class.  Adds functionality
 *  of blocking reads (producer/consumer) method.  Where multiple threads are blocked
 *  on read a round robin approach is created.  If no threads are blocked and event
 *  semaphores have been set, an event semaphore will be posted.  If the semaphore 
 *  does not have a listener the message will be lost.
 */

public class Queue
extends List
{
   private final int DEFAULT_SIZE = 50;

   boolean    _notify;
   EventSem[] _notifySems;
   int        _notifyCount;
   int        _notifyRobin;

   int      _waiting;
   EventSem _waitingSem;

   public Queue()
   {
      _notify = false;
      _notifySems = new EventSem[ DEFAULT_SIZE ];
      _notifyCount = 0;
      _notifyRobin = 0;

      _waiting = 0;
      _waitingSem = new EventSem();
   }

   /**
    * write an object to the queue and wake up a thread is any are blocked on read.
    */

   public void write( Object m )
   {
      synchronized ( this )
      {
          add( m );
          
          if ( _waiting > 0 ) 
          {
             _waitingSem.post();
             return;
          }

          if ( _notify )
          {
             int currentSem = _notifyRobin;
	     while ( _notifySems[ _notifyRobin ].query() > 0 ) 
             {
                _notifyRobin++;
                if (_notifyRobin >= _notifyCount ) _notifyRobin = 0;

                // This is dangerous if a problem occurs and we think all the semaphores have
                // been posted.  We are throwing the message on the queue and forgetting about it.
                if (_notifyRobin == currentSem ) return;  
             }

             _notifySems[ _notifyRobin ].post();
             //_notifyRobin++;
             return;
          }
      }
   }

   /**
    *  read an item from the Queue.  Block can be set to true to make the reader
    *  block if no messages are available.
    */

   public Object read( boolean block )
   {
      synchronized ( this )
      {
         if ( isEmpty() )
         {
            if ( !block )
               return null;

            _waiting++;
            try
            {
               while ( isEmpty() ) _waitingSem.wait();
            }
            catch( InterruptedException ex )
            {
              return null;
            }
            _waitingSem.reset();
            _waiting--;
         }
         return remove();
      }
   }

   /**
    *  set a semaphore to be posted when a message is placed on the queue.
    */

   public void setEventSem( EventSem sem )
   {
      synchronized( this )
      {
          _notifySems[ _notifyCount ] = sem;
          _notifyCount++;
          _notify = true;
      }
   }

}